package com.study.flink.batch;

import org.apache.flink.api.common.functions.MapPartitionFunction;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.operators.DistinctOperator;
import org.apache.flink.api.java.operators.MapPartitionOperator;
import org.apache.flink.util.Collector;

import java.util.ArrayList;
import java.util.Iterator;

/**
 * @author stephen
 * @create 2019-01-28 21:05
 * @since 1.0.0
 */
public class BatchDemo {

    public static void main(String[] args) throws Exception {
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

        ArrayList<String> data = new ArrayList<>();
        data.add("hello world");
        data.add("hello china");

        DataSource<String> source = env.fromCollection(data);

        MapPartitionOperator<String, String> mapPartitionData = source.mapPartition(new MapPartitionFunction<String, String>() {
            @Override
            public void mapPartition(Iterable<String> values, Collector<String> out) throws Exception {
                // 一个分区的数据都在这里处理
                // values中保存了一个分区的数据
                // 一般map需要操作第三方连接时使用
                Iterator<String> iterator = values.iterator();
                while (iterator.hasNext()) {
                    String s = iterator.next();
                    String[] split = s.split("\\W+");
                    for (String word : split) {
                        out.collect(word);
                    }
                }
            }
        });

        // 去重
        DistinctOperator<String> result = mapPartitionData.distinct();

        result.print();
    }
}
